package com.bwie.service.impl;

import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateTime;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bwie.mapper.TbDictMapper;
import com.bwie.mapper.TbPacketRecordMapper;
import com.bwie.poi.ExcelUtil;
import com.bwie.pojo.TbDict;
import com.bwie.pojo.TbPacketRecord;
import com.bwie.pojo.TbPolicy;
import com.bwie.policy.RedPacketConText;
import com.bwie.service.TbPolicyService;
import com.bwie.mapper.TbPolicyMapper;
import com.bwie.utils.Result;
import com.bwie.vo.PacketVo;
import com.bwie.vo.RecordPageVo;
import com.bwie.vo.RedInfoVo;
import com.bwie.vo.TimeInfoVo;
import io.jsonwebtoken.Claims;
import io.jsonwebtoken.Jwts;

import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;

import org.springframework.stereotype.Service;

import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import java.util.*;
import java.util.stream.Collectors;

/**
* @author 28242
* @description 针对表【tb_policy(红包策略表)】的数据库操作Service实现
* @createDate 2025-06-07 11:00:06
*/
@Service
public class TbPolicyServiceImpl extends ServiceImpl<TbPolicyMapper, TbPolicy>
    implements TbPolicyService{

    @Autowired
    TbDictMapper tbDictMapper;

    @Autowired
    TbPolicyMapper tbPolicyMapper;

    @Autowired
    StringRedisTemplate redisTemplate;

    @Autowired
    HttpServletRequest request;

    @Autowired
    HttpServletResponse response;

    @Autowired
    TbPacketRecordMapper tbPacketRecordMapper;


    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;




    @Autowired
    KafkaTemplate kafkaTemplate;


    @Override
    public Result getPacketList(PacketVo pageInfoVo) {

        //分页参数
        Page<TbPolicy> page = new Page<>(pageInfoVo.getPageNum(),pageInfoVo.getPageSize());

        QueryWrapper<TbPolicy> wrapper = new QueryWrapper<>();
        if (StringUtils.isNotEmpty(pageInfoVo.getName())){
            wrapper.lambda().like(TbPolicy::getPolicyName,pageInfoVo.getName());
        }
        if (pageInfoVo.getStatus()!=null){
            wrapper.lambda().eq(TbPolicy::getPacketEnable,pageInfoVo.getStatus());
        }


        //排序
        wrapper.lambda().orderByDesc(TbPolicy::getCreatedTime);

        //查询
        Page<TbPolicy> tbPolicyPage = tbPolicyMapper.selectPage(page, wrapper);


        //将条件查询的信息存入redis作为缓存
        String token = request.getHeader("token");
        Claims body = Jwts.parser().setSigningKey("123456").parseClaimsJws(token).getBody();
        Integer userId = body.get("userId", Integer.class);

        String key="policy-"+userId;
        redisTemplate.opsForValue().set(key, JSON.toJSONString(tbPolicyPage));

        return Result.ok().setData(tbPolicyPage);
    }

    @Override
    public Result getDictList(Integer typeId) {

        String key1="dict-"+typeId;
        Boolean aBoolean = redisTemplate.hasKey(key1);
        if (aBoolean){
            //查询字典
            List<Object> values = redisTemplate.opsForHash().values(key1);

            List<TbDict> tbDictList = values.stream().map(item -> {
                TbDict tbDict = JSON.parseObject(item.toString(), TbDict.class);
                return tbDict;
            }).collect(Collectors.toList());
            return Result.ok().setData(tbDictList);
        }else {
            //查询数据库
            List<TbDict> tbDicts = tbDictMapper.selectList(
                    new QueryWrapper<TbDict>().lambda().eq(TbDict::getDictType, typeId)
            );

            for (TbDict tbDict : tbDicts) {
                String key="dict-"+typeId;
                String key2=tbDict.getDictValue().toString();
                String value= JSON.toJSONString(tbDict);
                redisTemplate.opsForHash().put(key,key2,value);
            }

            return Result.ok().setData(tbDicts);
        }

    }

    @Override
    public void getExcele(Integer userId) {
        String key="policy-"+userId;
        String redispage = redisTemplate.opsForValue().get(key);

        List<TbPolicy> tbPolicies = JSON.parseArray(redispage, TbPolicy.class);

        ExcelUtil<TbPolicy> excelUtil = new ExcelUtil<>(TbPolicy.class);

        excelUtil.exportExcel(response,tbPolicies,"红包策略列表");


    }

    @Override
    public Result addPolicy(TbPolicy tbPolicy) {

        long beginTime = tbPolicy.getBeginTime().getTime();
        long endTime = tbPolicy.getEndTime().getTime();

        if ((endTime-beginTime)<1000*60*60*24){
            return Result.error().setMsg("策略时间应该在生效时间一天后");
        }

        tbPolicy.setLeftAmount(tbPolicy.getPacketAmount());

        tbPolicyMapper.insert(tbPolicy);


        return Result.ok();
    }

    @Override
    public Result grabRedPacket(Integer id) {

        //获取红包信息
        TbPolicy tbPolicy = tbPolicyMapper.selectById(id);
        if (tbPolicy==null){
            return Result.error().setMsg("红包不存在");
        }

        String token = request.getHeader("token");
        if (token==null){
            return Result.error().setMsg("请重新登录");
        }
        Claims body = Jwts.parser().setSigningKey("123456").parseClaimsJws(token).getBody();
        Integer userId = body.get("userId", Integer.class);


        //调用上下文获取红包策略
        //使用红包策略抢红包
        RedPacketConText redPacketConText = new RedPacketConText();
        redPacketConText.setRedPolicy(tbPolicy,userId,kafkaTemplate,redisTemplate);
        Result result = redPacketConText.doGrabRedPacket();



        return result;
    }

    @Override
    public Result getRecordList(RecordPageVo recordPageVo) {

        Integer pageNum=0;
        if (recordPageVo.getPageNum()>0){
            pageNum=recordPageVo.getPageNum()-1;
        }
        PageRequest pageRequest = PageRequest.of(pageNum, recordPageVo.getPageSize());

        BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

        if (StringUtils.isNotEmpty(recordPageVo.getKeyWord())){
            boolQueryBuilder.must(
                    QueryBuilders.matchQuery("policyName",recordPageVo.getKeyWord())
            );
        }

        NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
        nativeSearchQueryBuilder.withPageable(pageRequest).withQuery(boolQueryBuilder);

        SearchHits<TbPacketRecord> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), TbPacketRecord.class);

        long total = searchHits.getTotalHits();

        ArrayList<TbPacketRecord> records = new ArrayList<>();
        for (SearchHit<TbPacketRecord> searchHit : searchHits.getSearchHits()) {

            TbPacketRecord tbPacketRecord = searchHit.getContent();

            records.add(tbPacketRecord);
        }

        if (total==0){

            Page<TbPacketRecord> page = new Page<>(recordPageVo.getPageNum(),recordPageVo.getPageSize());
            Page<TbPacketRecord> recordList = tbPacketRecordMapper.getRecordList(page, recordPageVo.getKeyWord());
            return Result.ok().setData(recordList);


        }else {
            HashMap<String, Object> map = new HashMap<>();

            map.put("total",total);
            map.put("records",records);
            return Result.ok().setData(map);
        }

    }

    @Override
    public Result getEndTime(TimeInfoVo timeInfoVo) {

        Date beginTime = timeInfoVo.getBeginTime();
        Date endTime = new Date();
        switch (timeInfoVo.getType()){
            case 1://周
            {
                DateTime dateTime = DateTime.of(beginTime);
                DateTime offset = dateTime.offset(DateField.WEEK_OF_YEAR, 1);
                endTime = offset.toJdkDate();
            }

                break;
            case 2://天
            {
                DateTime dateTime = DateTime.of(beginTime);
                DateTime endDateTime = dateTime.offset(DateField.DAY_OF_YEAR, 1);
                endTime=endDateTime.toJdkDate();
            }

                break;
            case 3://月
            {
                DateTime dateTime = DateTime.of(beginTime);
                DateTime endDateTime = dateTime.offset(DateField.MONTH, 1);
                endTime=endDateTime.toJdkDate();
            }

                break;
            case 4://总次数
                break;
            default:
                break;
        }
        String s = DateTime.of(endTime).toString("yyyy-MM-dd HH:mm:ss");


        return Result.ok().setData(s);
    }

    @Override
    public Result getRedRecordInfo(RedInfoVo redInfoVo) {

        Page<TbPacketRecord> page = new Page<>(redInfoVo.getPageNum(),redInfoVo.getPageSize());

        Page<TbPacketRecord> tbPacketRecordPage = tbPacketRecordMapper.selectPageInfo(page,redInfoVo.getTypeId());

        return Result.ok().setData(tbPacketRecordPage);
    }

    @KafkaListener(topics = "topic-packet")
    public void rectRedPacket(String message, Acknowledgment acknowledgment){

        System.out.println("收到抢红包的消息");

        TbPacketRecord tbPacketRecord = JSON.parseObject(message, TbPacketRecord.class);
        Long msgId = tbPacketRecord.getMsgId();

        //判断消息是否重复
        String key="msg-"+msgId;
        Boolean aBoolean = redisTemplate.hasKey(key);
        if (!aBoolean){
            System.out.println("消息已经被消费过了");
            acknowledgment.acknowledge();
            return;
        }

        //消费消息
            TbPolicy tbPolicy = tbPolicyMapper.selectById(tbPacketRecord.getPolicyId());
            tbPolicy.setLeftAmount(tbPolicy.getPacketAmount()-tbPacketRecord.getPacketAmount());

            tbPolicyMapper.updateById(tbPolicy);
            tbPacketRecordMapper.insert(tbPacketRecord);


            saveRecordToEs(tbPacketRecord);

            //手动确认消息
            acknowledgment.acknowledge();
            redisTemplate.delete(key);

    }

    public void saveRecordToEs(TbPacketRecord tbPacketRecord){

        TbPacketRecord tbPacketRecord1 = tbPacketRecordMapper.selectEsRecord(tbPacketRecord.getRecordId());

        IndexOperations indexOps = elasticsearchRestTemplate.indexOps(TbPacketRecord.class);

        boolean exists = indexOps.exists();
        if (!exists){
            indexOps.create();
            Document mapping = indexOps.createMapping();
            indexOps.putMapping(mapping);
        }

        elasticsearchRestTemplate.save(tbPacketRecord1);


    }
}




